package com.mtiiot.fstcpserver.mq.kafka.insplatform;


import com.mtiiot.fstcpserver.mq.kafka.pool.CreateKafkaProducerPool;
import com.mtiiot.fstcpserver.mq.kafka.pool.KafkaProducerApp;
import kafka.producer.KeyedMessage;
import org.apache.commons.pool2.impl.GenericObjectPool;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * Created by xyr on 2016/7/26.
 */
public class InstruProducer {
    private static  GenericObjectPool<KafkaProducerApp> pool = null;
    //private static KafkaProducerApp producer = null; // 2016.09.01 注释

    /*static{
        CreateKafkaProducerPool e = new CreateKafkaProducerPool();
        pool = e.createKafkaProducerPool();
        try {
            producer = (KafkaProducerApp)pool.borrowObject();
        } catch (Exception e1) {
            e1.printStackTrace();
        }
    }*/

    // 2016.09.01
    static{
        // 创建线程池
        CreateKafkaProducerPool e = new CreateKafkaProducerPool();
        pool = e.createKafkaProducerPool();
    }

    /**
     * 在线程池中创建一个producer
     * */
    public static KafkaProducerApp createProducer() throws Exception {
        KafkaProducerApp producer = null;
        producer = (KafkaProducerApp)pool.borrowObject();
        return producer;
    }

    /**
     * 归还producer对象到池
     * */
    public static void returnProducer(KafkaProducerApp producer){
        pool.returnObject(producer);
    }


    /*protected void finalize() throws Throwable {
        super.finalize();
        this.pool.returnObject(producer);
        this.pool.close();
    }*/

    /**
     * 发送单条消息
     * */
    public static void sendData(String code, String msgKey, String msg) throws Exception {
        // 先从对象池中拿到
        KafkaProducerApp producer =createProducer();
        producer.sendData(code, msgKey, msg);
        // 归还
        returnProducer(producer);
    }

    /**
     * 发送批量消息
     * */
    /*public static void sendDatas(List<KeyedMessage<String,String>> list) throws Exception{
        // 先从对象池中拿到
        KafkaProducerApp producer =createProducer();
        producer.sendDatas(list);
        // 归还
        returnProducer(producer);
    }*/
    public static void sendDatas(String code,Map<String,String> map) throws Exception{
        // 先从对象池中拿到
        KafkaProducerApp producer =createProducer();
        List<KeyedMessage<String,String>> list = new ArrayList<KeyedMessage<String, String>>();
        for (Map.Entry<String,String> m : map.entrySet()){
            KeyedMessage message = new KeyedMessage(code,m.getKey() , m.getValue());
            list.add(message);
        }
        producer.sendDatas(list);
        // 归还
        returnProducer(producer);
    }


}
